package org.nbict.iot.trash;

import com.alibaba.fastjson.JSONObject;
import com.mongodb.BasicDBObject;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.nbict.iot.persist.pool.DbRepository;
import org.nbict.iot.persist.repo.MongoRepo;

import java.util.HashMap;
import java.util.Map;

/**
 * S0
 * 报文头部动作分发
 * Created by songseven on 18/6/13.
 */
@Deprecated
public class S0Bolt extends BaseBasicBolt {

    private MongoRepo repo;

    public void prepare(Map stormConf, TopologyContext context) {
        repo = (MongoRepo) DbRepository.MONGO.getInstance();
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector
            collector) {
        JSONObject o = (JSONObject)input.getValueByField("parsed");
        JSONObject header = o.getJSONObject("car_packet_header");
        JSONObject body = (JSONObject)o.getJSONObject("car_packet_body");
        String companyId = o.getString("company_id");
        Long timestamp = body.getLong("timestamp"); //时间戳

        int cmd = header.getInteger("header_cmd");
        String vin = header.getString("car_vin");
        //实时信息上报或者补发信息上报
        Map history = new HashMap();
        history.put("platform_id", companyId);
        history.put("car_vin", vin);
        history.put("action", cmd);
        history.put("timestamp", timestamp);
        history.put("info", body);
        repo.insert("car_history", new BasicDBObject(history));

        if (cmd == 2 || cmd == 3) {
            collector.emit("S0-2", new Values(o));
        } else {
            collector.emit("S0-1", new Values(o));
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("S0-1", new Fields("parsed")); //login and logout
        declarer.declareStream("S0-2", new Fields("parsed")); //payload
    }
}
